package kq

import (
	"context"
	"encoding/json"
	"fmt"
	"strconv"
	"strings"

	"mall/common/cachex"
	"mall/common/globalkey"
	"mall/common/mqueuex"
	"mall/common/utils/pointy"
	enum "mall/service/backend/common/enum/user"
	"mall/service/backend/rmq/internal/svc"
	"mall/service/order/rpc/orderclient"
	"mall/service/product/rpc/productclient"
	"mall/service/promotion/rpc/promotionclient"
	"mall/service/user/rpc/userclient"

	"github.com/dtm-labs/client/dtmcli/dtmimp"
	"github.com/dtm-labs/client/dtmgrpc"
	"github.com/pkg/errors"
	"github.com/sirupsen/logrus"
	"github.com/zeromicro/go-zero/core/logx"
)

/**
 * Listening to the payment flow status change notification message queue
 */
type DeferedUpdateTradeStateMq struct {
	ctx    context.Context
	svcCtx *svc.ServiceContext
}

func NewDeferedUpdateTradeStateMq(ctx context.Context, svcCtx *svc.ServiceContext) *DeferedUpdateTradeStateMq {
	return &DeferedUpdateTradeStateMq{
		ctx:    ctx,
		svcCtx: svcCtx,
	}
}

func (l *DeferedUpdateTradeStateMq) Consume(_, val string) error {
	logrus.Info("DeferedUpdateTradeStateMq Consume exec")
	// 解析数据
	var message mqueuex.DeferedUpdateTradeStateMessage
	if err := json.Unmarshal([]byte(val), &message); err != nil {
		// logx.WithContext(l.ctx).Error("DeferedUpdateTradeStateMq.Consume Unmarshal err : %v , val : %s", err, val)
		logx.Errorw("DeferedUpdateTradeStateMq.Consume json.Unmarshal err", logx.Field("detail", err.Error()))
		return err
	}

	// 执行业务..
	if err := l.execService(&message); err != nil {
		// logx.WithContext(l.ctx).Error("DeferedUpdateTradeStateMq.execService err : %v , val : %s , message: %+v", err, val, message)
		logx.Errorw("DeferedUpdateTradeStateMq.execService err", logx.Field("detail", err.Error()),
			logx.Field("val", val), logx.Field("message", fmt.Sprintf("%+v", message)))
		return err
	}

	return nil
}

// Verify and update relevant flow data
func (l *DeferedUpdateTradeStateMq) execService(message *mqueuex.DeferedUpdateTradeStateMessage) error {
	logrus.Info("DeferedUpdateTradeStateMq Consume execService")
	if message.OrderSn == nil {
		return errors.Wrapf(OrderUpdateTradeStateError, "Failed to get paid order due to invalid param, message: %+v ", message)
	}

	order, err := l.svcCtx.OrderRpc.GetOrderBySn(l.ctx, &orderclient.GetOrderBySnReq{
		Sn:            message.OrderSn,
		WithSku:       pointy.GetPointer(uint32(1)),
		WithPromotion: pointy.GetPointer(uint32(1)),
		WithPayment:   pointy.GetPointer(uint32(1)),
	})
	// logrus.Info(fmt.Sprintf("DeferedUpdateTradeStateMq execService order: %+v", order))
	if err != nil {
		return errors.Wrapf(OrderUpdateTradeStateError, "Failed to get paid order err: %v , message: %+v ", err, message)
	}

	if order.Id == nil || order.OrderType == nil || order.TradeState == nil || order.UserId == nil || order.Gems == nil {
		return errors.Wrapf(OrderUpdateTradeStateError, "Failed to get valid data of paid order: %+v ", order)
	}

	if *order.OrderType != globalkey.OrderTypeVirtual && *order.TradeState == globalkey.OrderTradeStateNotPay {
		var paymentId *uint64
		if len(order.Payments) > 0 {
			payment := order.Payments[0]
			paymentId = payment.Id
		}
		var stocks []*productclient.ProductVariationStock
		if len(order.Skus) > 0 {
			for _, v := range order.Skus {
				if v.SkuId != nil && v.Quantity != nil && *v.SkuId > 0 && *v.Quantity > 0 {
					stocks = append(stocks, &productclient.ProductVariationStock{
						Id:    *v.SkuId,
						Stock: *v.Quantity,
					})
				}
			}
		}
		var promotionLogs []*promotionclient.PromotionLog
		if len(order.Promotions) > 0 {
			for _, v := range order.Promotions {
				promotionLogs = append(promotionLogs, &promotionclient.PromotionLog{
					ActivityId:       v.ActivityId,
					ActivityRuleId:   v.ActivityRuleId,
					CouponTemplateId: v.CouponTemplateId,
					CouponId:         v.CouponId,
				})
			}
		}
		var amount *string
		if order.Gems != nil && *order.Gems > 0 {
			amount = pointy.GetPointer(strconv.FormatInt(int64(*order.Gems), 10))
		}

		// Update the flow status.
		err = l.updateTradeStateDtmSaga(
			order.Id,
			pointy.GetPointer(globalkey.OrderTradeStatePaid),
			order.TradeState,
			paymentId,
			order.UserId,
			amount,
			order.Sn,
			stocks,
			promotionLogs,
		)
		if err != nil {
			return errors.Wrapf(OrderUpdateTradeStateError, "update order tradeState failed, err: %v , message: %+v", err, message)
		}
	}

	return nil
}

func (l *DeferedUpdateTradeStateMq) updateTradeStateDtmSaga(
	orderId *uint64,
	tradeState *uint32,
	tradeStateRevert *uint32,
	paymentId *uint64,
	userId *uint64,
	amount *string,
	orderSn *string,
	stocks []*productclient.ProductVariationStock,
	promotionLogs []*promotionclient.PromotionLog,
) error {
	// 获取 OrderRpc BuildTarget
	orderRpcBusiServer, err := l.svcCtx.Config.OrderRpc.BuildTarget()
	if err != nil {
		return err
	}

	// 获取 UserRpc BuildTarget
	userRpcBusiServer, err := l.svcCtx.Config.UserRpc.BuildTarget()
	if err != nil {
		return err
	}

	// 获取 ProductRpc BuildTarget
	productRpcBusiServer, err := l.svcCtx.Config.ProductRpc.BuildTarget()
	if err != nil {
		return err
	}

	// 获取 PromotionRpc BuildTarget
	promotionRpcBusiServer, err := l.svcCtx.Config.PromotionRpc.BuildTarget()
	if err != nil {
		return err
	}

	//orderRpcBusiServer = "http://127.0.0.1:9030"
	//userRpcBusiServer = "http://127.0.0.1:9001"
	//productRpcBusiServer = "http://127.0.0.1:9020"
	//promotionRpcBusiServer = "http://127.0.0.1:9032"
	// orderRpcBusiServer = "127.0.0.1:9030"
	// userRpcBusiServer = "127.0.0.1:9001"
	// productRpcBusiServer = "127.0.0.1:9020"
	// promotionRpcBusiServer = "127.0.0.1:9032"

	// DefaultGrpcServer default url for grpc server.
	defaultGrpcServer := l.svcCtx.Config.DefaultGrpcServer
	// defaultGrpcServer := "193.112.95.184:36790"

	// 创建一个gid
	gid := dtmgrpc.MustGenGid(defaultGrpcServer)

	// 创建一个saga协议的事务
	saga := dtmgrpc.NewSagaGrpc(defaultGrpcServer, gid).
		Add(
			orderRpcBusiServer+"/order.Order/UpdateTradeStateRaw",
			orderRpcBusiServer+"/order.Order/UpdateTradeStateRevert",
			&orderclient.UpdateTradeStateReq{
				Id:               orderId,
				TradeState:       tradeState,
				TradeStateRevert: tradeStateRevert,
				PaymentId:        paymentId,
			}).
		Add(
			userRpcBusiServer+"/user.User/UpdateUserAssetRaw",
			userRpcBusiServer+"/user.User/UpdateUserAssetRevert",
			&userclient.UpdateUserAssetReq{
				UserId:       userId,
				AssetType:    pointy.GetPointer(enum.UserAssetTypeGem),            // 宝石
				ChangeType:   pointy.GetPointer(enum.UserAssetChangeTypeDecrease), // 减少
				ChangeAmount: amount,
				SourceType:   pointy.GetPointer(enum.UserAssetSourceTypeConsume), // 消费
				SourceId:     orderId,
				SourceDesc:   orderSn,
			}).
		Add(
			productRpcBusiServer+"/product.Product/UpdateStockRaw",
			productRpcBusiServer+"/product.Product/UpdateStockRevert",
			&productclient.UpdateStockReq{
				Data: stocks,
			}).
		Add(
			promotionRpcBusiServer+"/promotion.Promotion/UpdatePromotionLogRaw",
			promotionRpcBusiServer+"/promotion.Promotion/UpdatePromotionLogRevert",
			&promotionclient.UpdatePromotionLogReq{
				UserId:       userId,
				OrderId:      orderId,
				Status:       pointy.GetPointer(uint32(globalkey.StatusUsed)),
				StatusRevert: pointy.GetPointer(uint32(globalkey.StatusStarted)),
				Logs:         promotionLogs,
			})

	//扣减顺手带一件的积分
	orderKey := l.getOrderFrozenKey(*orderId)
	orderValue, err := l.svcCtx.BizRedis.GetCtx(context.Background(), orderKey)
	if err != nil {
		return err
	}

	fmt.Println("orderValue", orderValue)
	a, b, c, d, point := l.parseOrderValue(orderValue)
	fmt.Println("a, b, c, d, point ", a, b, c, d, point)

	if len(strings.TrimSpace(point)) > 0 {
		saga.Add(
			userRpcBusiServer+"/user.User/UpdateUserAssetRaw",
			userRpcBusiServer+"/user.User/UpdateUserAssetRevert",
			&userclient.UpdateUserAssetReq{
				UserId:       userId,
				AssetType:    pointy.GetPointer(enum.UserAssetTypePoint),          // 积分
				ChangeType:   pointy.GetPointer(enum.UserAssetChangeTypeDecrease), // 减少
				ChangeAmount: pointy.GetPointer(point),
				SourceType:   pointy.GetPointer(enum.UserAssetSourceTypeConsume), // 消费
				SourceId:     orderId,
				SourceDesc:   orderSn,
			})
	}

	// 事务提交
	err = saga.Submit()
	if err != nil {
		logrus.Info(fmt.Sprintf("DeferedUpdateTradeStateMq updateTradeStateDtmSaga err: %+v", err))
	}
	dtmimp.FatalIfError(err)
	if err != nil {
		return err
	}

	return nil
}

func (l *DeferedUpdateTradeStateMq) parseOrderValue(value string) (string, string, string, string, string) {
	//value = "优惠券id:活动id:logsids:goodsId:point"
	couponIds := strings.Split(value, ":")[0]
	activityIds := strings.Split(value, ":")[1]
	logsIds := strings.Split(value, ":")[2]
	goodsId := strings.Split(value, ":")[3]
	point := strings.Split(value, ":")[4]
	return couponIds, activityIds, logsIds, goodsId, point
}

func (l *DeferedUpdateTradeStateMq) getOrderFrozenKey(oid uint64) string {
	return fmt.Sprintf("{%s}:%d", cachex.OrderFrozen, oid)
}

//# 查看活跃的消费者组
//docker exec -it 5bb54468e662 kafka-consumer-groups \
//--bootstrap-server localhost:9092 \
//--list
//
//docker exec -it kafka kafka-topics \
//--bootstrap-server localhost:9092 \
//--create \
//--topic orders \
//--partitions 3 \
//--replication-factor 1
//
//docker exec -it 5bb54468e662 kafka-topics \
//--bootstrap-server localhost:9092 \
//--topic test \
//--describe
//
//
//docker exec -it  5bb54468e662 kafka-topics \
//--bootstrap-server localhost:9092 \
//--create \
//--topic defered-update-tradestate-topic \
//--partitions 3 \
//--replication-factor 1
//
//docker exec -it 5bb54468e662 kafka-topics \
//--bootstrap-server localhost:9092 \
//--delete \
//--topic defered-update-tradestate-topic
//
//echo "Order ID: 123" | docker exec -i 5bb54468e662 kafka-console-producer \
//--bootstrap-server localhost:9092 \
//--topic defered-update-tradestate-topic
//
//docker exec -it 5bb54468e662 kafka-console-consumer \
//--bootstrap-server localhost:9092 \
//--topic defered-update-tradestate-topic \
//--from-beginning
//
//docker exec -it 5bb54468e662 bash
//
//kafka-consumer-groups.sh --bootstrap-server localhost:9092 --list
//
//docker exec 5bb54468e662  /opt/bitnami/kafka/bin/kafka-consumer-groups.sh \
//--bootstrap-server localhost:9092 --list
//
//docker exec  5bb54468e662 ls -l /opt/bitnami/kafka/bin/ | grep kafka-consumer-groups
//
//- KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://193.112.95.184:9092
//- KAFKA_CFG_LISTENERS=PLAINTEXT://0.0.0.0:9092
//
//kafka-console-producer.sh \
//--bootstrap-server 193.112.95.184:9092 \
//--topic test-topic
//
//telnet 193.112.95.184 9092
//
//docker exec 5bb54468e662 cat /opt/bitnami/kafka/config/server.properties | grep advertised
//
//docker exec -it 5bb54468e662 \
///opt/bitnami/kafka/bin/kafka-console-producer.sh \
//--bootstrap-server localhost:9092 \
//--topic test-topic
//
//kafka-console-producer.sh \
//--bootstrap-server 193.112.95.184:9092 \
//--topic test-topic
